package com.ys.utils.redis.message;

import com.alibaba.fastjson.JSON;
import com.ys.utils.base.j2se.Logger;
import com.ys.utils.redis.Redis;
import com.ys.web.sys.redisMessage.model.RedisMessageService;
import com.ys.web.sys.redisMessage.model.SysRedisMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * producer/consumer消息队列父类
 */
@Slf4j
public abstract class QueueRedisMessageListener extends Thread {

    @Autowired
    private RedisMessageService redisMessageService;

    public abstract String getQueue();

    protected abstract boolean onMessage(RedisMessage message);

    protected boolean handleSuccess() {
        int handle = redisMessageService.handle(2, threadLocal.get().getId());
        return handle == 1;
    }
    protected boolean handleFailure() {
        int handle = redisMessageService.handle(3, threadLocal.get().getId());
        return handle == 1;
    }
    protected boolean handleFailure(String remark) {
        int handle = redisMessageService.handleErr(threadLocal.get().getId(), remark);
        return handle == 1;
    }

    public void onMessage(RedisMessage redisMessage, String queue) {
        SysRedisMessage sysRedisMessage = this.saveMessage(redisMessage);
        threadLocal.set(sysRedisMessage);
        try {
            boolean handlerResult = this.onMessage(redisMessage);
            redisMessageService.handle(handlerResult ? 2 : 3, sysRedisMessage.getId());
        } catch (Exception e) {
            Logger.error("redis队列处理因异常导致失败！队列名称:"+this.getQueue(), e);
            redisMessageService.handleErr(sysRedisMessage.getId(), e.getMessage());
        }
    }

    private SysRedisMessage saveMessage(RedisMessage message) {
        SysRedisMessage sysRedisMessage = new SysRedisMessage();
        sysRedisMessage.setChannel(this.getQueue());
        sysRedisMessage.setMessage(JSON.toJSONString(message));
        redisMessageService.create(sysRedisMessage);
        return sysRedisMessage;
    }

    private ThreadLocal<SysRedisMessage> threadLocal = new ThreadLocal<>();

    @Override
    public void run() {
        do {
            String o = null;
            try {
                o = Redis.list.rightPop(this.getQueue(), 2);
            } catch (Exception ignored) {}
            if (o == null) {
                continue;
            }
            try {
                this.onMessage(JSON.parseObject(o,RedisMessage.class), this.getQueue());
            } catch (Exception e) {
                Logger.error("redis消息处理失败！", e);
            }
        } while (!Thread.currentThread().isInterrupted());
    }

    @PostConstruct
    public void init() {
        this.setName("queue-" + this.getQueue());
//        this.start();
    }

    @PreDestroy
    public void onClose() {
        this.interrupt();
    }
}
